package com.yootk.rsocket.server.handler;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.util.DefaultPayload;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * RSocket处理类
 */
@Slf4j
public class MessageRSocketHandler implements RSocket {

    // 无响应
    @Override
    public Mono<Void> fireAndForget(Payload payload) {
        String message = payload.getDataUtf8();     // 获取数据
        log.info("【FireAndForget】接收请求数据：{}", message);
        return Mono.empty();    // 返回空消息
    }

    // 处理请求与响应
    @Override
    public Mono<Payload> requestResponse(Payload payload) {
        String message = payload.getDataUtf8();     // 获取数据
        log.info("【RequestAndResponse】接收请求数据：{}", message);
        return Mono.just(DefaultPayload.create("【ECHO】" + message));    // 数据响应
    }

    // 处理数据流
    @Override
    public Flux<Payload> requestStream(Payload payload) {
        String message = payload.getDataUtf8();     // 获取数据
        log.info("【RequestStream】接收请求数据：{}", message);
        return Flux.fromStream(message.chars()                      // 字符串转字符
                .mapToObj(c -> Character.toUpperCase((char) c))     // 字符转大写
                .map(Object::toString)                              // 字符转字符串
                .map(DefaultPayload::create));
    }

    // 双向流
    @Override
    public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
        return Flux.from(payloads).map(Payload::getDataUtf8).map(msg -> {
            log.info("【RequestChannel】接收请求数据：{}", msg);
            return msg;     // 返回数据
        }).map(DefaultPayload::create);
    }
}
